package bch.kafka.integrate.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * topic: 生产者往 topic 里发送消息，消费者从 topic 里消费消息
 * partition: Topic 内部分片，每个 topic 可以有多个 partition
 * group: 消费者组，一个组内可以有多个消费者，一个组内的消费者会负载均衡消费 topic 中的消息
 * offset: 消费者消费消息的位置，每个消费者消费到的消息都有一个 offset，offset 是一个整数，表示消息在 partition 中的位置
 */
@Component
@Slf4j
public class KafkaConsumer {

    @KafkaListener(topics = "topic-a", groupId = "group-a")
    public void listen1(String message, Acknowledgment ack) {
        log.info("topic-a-group-a收到消息: {}", message);
        ack.acknowledge();
    }

    @KafkaListener(topics = "topic-a", groupId = "group-b")
    public void listen2(String message, Acknowledgment ack) {
        log.info("topic-a-group-b收到消息: {}", message);
        ack.acknowledge();
    }



    @KafkaListener(topics = "topic-b", groupId = "group-a")
    public void listen4(String message, Acknowledgment ack) {
        log.info("topic-b-group-a收到消息: {}", message);
        ack.acknowledge();
    }

    @KafkaListener(topics = "topic-b", groupId = "group-b")
    public void listen5(String message, Acknowledgment ack) {
        log.info("topic-b-group-b收到消息: {}", message);
        ack.acknowledge();
    }

}
